package com.gaofans.vertx.boot.dbutils.session.impl;

import com.gaofans.vertx.boot.dbutils.session.ExecuteOperator;
import io.vertx.sqlclient.*;
import reactor.core.publisher.Mono;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

public class ExecuteOperatorImpl<S extends SqlClient> implements ExecuteOperator<S> {

    private S sqlClient;
    private List<List<Object>> batchParams;
    private List<Object> params;
    private String sql;

    ExecuteOperatorImpl(S sqlClient) {
        this.sqlClient = sqlClient;
    }

    @Override
    public ExecuteOperator<S> by(String sql) {
        this.sql = sql;
        return this;
    }

    @Override
    public ExecuteOperator<S> batch(List<Object>... params) {
        this.batchParams = Arrays.asList(params);
        return this;
    }

    @Override
    public ExecuteOperator<S> with(Object... params) {
        this.params = Arrays.asList(params);
        return this;
    }

    @Override
    public Mono<Void> execute() {
        if(params == null && params.size() != 0 && batchParams == null && batchParams.size() != 0){
            return Mono.create(sink -> {
                sqlClient.query(sql).execute(res -> {
                    if(res.succeeded()){
                        sink.success();
                    }else{
                        sink.error(res.cause());
                    }
                });
            });
        }else {
            return Mono.create(sink -> {
                PreparedQuery<RowSet<Row>> query = sqlClient.preparedQuery(sql);
                if(batchParams != null && batchParams.size() != 0){
                    List<Tuple> tuples = batchParams.stream().map(Tuple::wrap).collect(Collectors.toList());
                    query.executeBatch(tuples,res -> {
                        if(res.succeeded()){
                            sink.success();
                        }else{
                            sink.error(res.cause());
                        }
                    });
                }else{
                    sqlClient.preparedQuery(sql).execute(Tuple.wrap(new ArrayList<>(params)), res -> {
                        if(res.succeeded()){
                            sink.success();
                        }else{
                            sink.error(res.cause());
                        }
                    });
                }
            });
        }
    }
}
